// Copyright 2022 Huawei Cloud Computing Technology Co., Ltd.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include <unistd.h>
#include <string>
#include <sstream>
#include "CasSocket.h"
#include "CasMsg.h"
#include "CasBuffer.h"
#include "CasLog.h"
#include "CasAppCtrlCmdUtils.h"
#include "CasHeartbeatController.h"

using namespace std;

CasHeartbeatController::CasHeartbeatController(CasStreamBuildSender *streamBuildSender)
{
    m_streamBuildSender = streamBuildSender;
    m_status = true;
    Init();
}

void CasHeartbeatController::Init()
{
    m_recvStream = new(std::nothrow) CasDataPipe(true);
    if (m_recvStream == nullptr) {
        ERR("Failed to new recv stream.");
    }
}

CasHeartbeatController::~CasHeartbeatController()
{
    delete m_recvStream;
    this->m_streamBuildSender = nullptr;
}

void CasHeartbeatController::StopHandle()
{
    m_status = false;
    m_recvStream->Exit();
}

void CasHeartbeatController::Handle(void *pPacket)
{
    m_recvStream->Handle(pPacket);
}

bool CasHeartbeatController::HeartBeatRequest(int timeout)
{
    std::lock_guard<std::mutex> lockGuard(this->m_lock);
    this->m_recvStream->Clear();
    map<string, string> parameters = { { KEY_COMMAND, CMD_HEARTBEAT_REQUEST } };
    string msg = CasAppCtrlCmdUtils::MakeCommand(parameters);
    return Send(msg) && WaitResponse(CMD_HEARTBEAT_RESPONSE, timeout);
}

bool CasHeartbeatController::Send(string msg)
{
    int ret = m_streamBuildSender->SendDataToServer(CasMsgType::HeartBeat, msg.c_str(), msg.size() + 1);
    if (ret != static_cast<int>(msg.size() + 1)) {
        ERR("Send ret is %d, expect is %d, %s", ret, (int)msg.size() + 1, msg.c_str());
        return false;
    }
    return true;
}

bool CasHeartbeatController::WaitResponse(string command, int timeout)
{
#if CONTROL_RESPONSE_ENABLED
    void *pPkt = nullptr;
    pPkt = m_recvStream->GetNextPktWaitFor(timeout);
    if (pPkt == nullptr) {
        if (!m_status) {
            ERR("Stop running");
            return true;
        }
        ERR("Wait response timeout");
        return false;
    }

    streamMsgHead *streamHead = (streamMsgHead *)pPkt;
    const char *receivedMsgString = (char *)((uint8_t *)pPkt + sizeof(streamMsgHead));

    map<string, string> parameters = CasAppCtrlCmdUtils::ParseCommand(receivedMsgString, streamHead->GetPayloadSize());
    if (parameters.find(KEY_COMMAND) == parameters.end()) {
        ERR("Invalid response %s.", CasAppCtrlCmdUtils::MakeCommand(parameters).c_str());
    }

    FreeBuffer(pPkt);
#endif
    return true;
}
